package com.lingshi.shopping_seckill_service.service;

import cn.hutool.bloomfilter.BitMapBloomFilter;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.IdUtil;
import com.alibaba.csp.sentinel.annotation.SentinelResource;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.lingshi.shopping.constant.RedisKey;
import com.lingshi.shopping.entity.CartGoods;
import com.lingshi.shopping.entity.Orders;
import com.lingshi.shopping.entity.SeckillGoods;
import com.lingshi.shopping.enums.ErrorEnums;
import com.lingshi.shopping.enums.OrderStatusEnums;
import com.lingshi.shopping.exception.BusException;
import com.lingshi.shopping.service.SecKillService;
import com.lingshi.shopping_seckill_service.mapper.SecKillGoodsMapper;
import com.lingshi.shopping_seckill_service.component.RedissonLock;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.scheduling.annotation.Scheduled;

import java.time.Duration;
import java.util.Date;
import java.util.List;
import java.util.Objects;

@Slf4j
@DubboService
public class SecKillServiceImpl implements SecKillService {

    @Autowired
    private SecKillGoodsMapper seckillGoodsMapper;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private BitMapBloomFilter bitMapBloomFilter;

    @Autowired
    private RedissonLock redissonLock;

    /**
     * 每4s查询一次数据库，更新redis中的秒杀商品数据
     * 条件为startTime < 当前时间 < endTime，库存大于0
     */
    @Scheduled(cron = "0/30 * * * * *")
    public void syncSecKillGoodsToRedis() {

        //查询redis的数据同步到mysql: 主要为了同步已经购买过的库存
        String key = RedisKey.SEC_KILL_GOODS;
        HashOperations<String, String, SeckillGoods> hos = redisTemplate.opsForHash();
        List<SeckillGoods> seckillGoodsList = hos.values(key);
        if (CollUtil.isNotEmpty(seckillGoodsList)) {
            for (SeckillGoods seckillGoods : seckillGoodsList) {
                seckillGoodsMapper.updateById(seckillGoods);
                System.out.println("seckillGoods = " + seckillGoods);
            }
        }


        //查询数据库数据同步redis
        LambdaQueryWrapper<SeckillGoods> queryWrapper = new LambdaQueryWrapper<>();
        Date now = new Date();
        queryWrapper.lt(SeckillGoods::getStartTime, now);
        queryWrapper.gt(SeckillGoods::getEndTime, now);
        queryWrapper.gt(SeckillGoods::getStockCount, 0);

        List<SeckillGoods> secKillGoods = seckillGoodsMapper.selectList(queryWrapper);

        //删除之前的所有秒杀删除（可能之前有过期）
        redisTemplate.delete(key);


        if (CollUtil.isNotEmpty(secKillGoods)) {
            //同步秒杀商品到Redis
            for (SeckillGoods secKillGood : secKillGoods) {
                hos.put(key, secKillGood.getGoodsId().toString(), secKillGood);
                //添加到布隆过滤器
                bitMapBloomFilter.add(secKillGood.getGoodsId().toString());
            }

        }

    }


    @Override
    @SentinelResource("findPageByRedis")
    public Page<SeckillGoods> findPageByRedis(int page, int size) {

        String key = RedisKey.SEC_KILL_GOODS;

        HashOperations<String, String, SeckillGoods> hos = redisTemplate.opsForHash();
        //Redis获取所有的秒杀商品
        List<SeckillGoods> seckillGoodsList = hos.values(key);

        Page<SeckillGoods> objectPage = new Page<>();

        if (CollUtil.isNotEmpty(seckillGoodsList)) {
            //计算数据其实位置
            int begin = (page - 1) * size;
            //计算数据结束位置
            int end = begin + size <= seckillGoodsList.size() ? begin + size : seckillGoodsList.size();

            //获取每一页数据
            List<SeckillGoods> pageData = seckillGoodsList.subList(begin, end);

            //设置分页数据
            objectPage.setRecords(pageData);
            //设置分页总数
            objectPage.setTotal(seckillGoodsList.size());
        }
        //线程睡眠500ms，模拟访问速度慢
        ThreadUtil.sleep(500);

        return objectPage;
    }

    @Override
    public SeckillGoods findSeckillGoodsByRedis(String goodsId) {

        //布隆过滤器
//        if (!bitMapBloomFilter.contains(goodsId)) {
//            log.info("=============商品id {}不存在=================", goodsId);
//            return null;
//        }


        String key = RedisKey.SEC_KILL_GOODS;
        HashOperations<String, String, SeckillGoods> hos = redisTemplate.opsForHash();
        //redis获取秒杀商品
        SeckillGoods seckillGoods = hos.get(key, goodsId);

        return seckillGoods;
    }

    @Override
    @SentinelResource(value = "findSecillGoodsByMySql", blockHandler = "mySqlBlockHandler")
    public SeckillGoods findSeckillGoodsByMySQL(String goodsId) {
        LambdaQueryWrapper<SeckillGoods> queryWrapper = new LambdaQueryWrapper<>();
        Date now = new Date();
        queryWrapper.lt(SeckillGoods::getStartTime, now);
        queryWrapper.gt(SeckillGoods::getEndTime, now);
        queryWrapper.gt(SeckillGoods::getStockCount, 0);
        queryWrapper.eq(SeckillGoods::getGoodsId, goodsId);
        SeckillGoods seckillGoods = seckillGoodsMapper.selectOne(queryWrapper);
        //同步到redis
        if (Objects.isNull(seckillGoods)) {
            addRedisSeckillGoods(seckillGoods);
        }
        return seckillGoods;
    }

    public SeckillGoods mySqlBlockHandler(String goodsId, BlockException e) {
        log.info("mySqlBlockHandler :{}", goodsId);
        return null;
    }


    @Override
    public Orders  createOrder(Orders orders) {
        //创建订单id
        String orderId = IdUtil.getSnowflakeNextIdStr();
        orders.setStatus(OrderStatusEnums.UNPAID.getCode());
        orders.setId(orderId);
        orders.setCreateTime(new Date());
        //设置过期时间30分钟
        DateTime expireDate = DateUtil.offsetMinute(new Date(), 30);
        orders.setExpire(expireDate);

        //获取购买的商品
        CartGoods cartGoods = orders.getCartGoods().get(0);
        //设置商品的订单id
        cartGoods.setOrderId(orderId);

        if (redissonLock.lock(cartGoods.getGoodId().toString(), 10000L)) {
            try {
                //查询redis商品
                SeckillGoods seckillGoodsByRedis = this.findSeckillGoodsByRedis(cartGoods.getGoodId().toString());


                if (Objects.isNull(seckillGoodsByRedis) || seckillGoodsByRedis.getStockCount() <= 0 || cartGoods.getNum() > seckillGoodsByRedis.getStockCount()) {
                    throw new BusException(ErrorEnums.GOODS_STOCK_ERROR);
                }

                //扣减库存
                seckillGoodsByRedis.setStockCount(seckillGoodsByRedis.getStockCount() - cartGoods.getNum());

                //睡眠100ms
                ThreadUtil.sleep(100);
                //购买以后 同步到redis
                String key = RedisKey.SEC_KILL_GOODS;
                HashOperations<String, String, SeckillGoods> hos = redisTemplate.opsForHash();
                hos.put(key, cartGoods.getGoodId().toString(), seckillGoodsByRedis);

                //下订单: 使用 string类型，订单id作为key，订单明细作为值
                redisTemplate.boundValueOps(orderId).set(orders, Duration.ofMinutes(1));
                /**
                 *
                 * redis可以监听key过期
                 * 但是：只能获取到获过期的key，不能获取到过期的key对应的值
                 * 解决方案：
                 *   设置一个订单id+后缀（前缀） 比订单万过期的和订单一一样的key
                 *  当订单过期时候，可以获取key
                 *  使用 key + 后缀(前缀) 获取晚过期的数据，可以得到商品数量，回退数量即可
                 */
                redisTemplate.boundValueOps(orderId + "_copy").set(orders, Duration.ofMinutes(2));
            } finally {
                redissonLock.unlock(cartGoods.getGoodId().toString());
            }
        }

        return orders;
    }

    @Override
    public Orders findOrder(String id) {
//        Orders orders = (Orders)redisTemplate.boundValueOps(id).get();
        ValueOperations<String, Orders> vos = redisTemplate.opsForValue();
        return vos.get(id);
    }

    @Override
    public Orders pay(String orderId) {

        Orders orders = findOrder(orderId);
        if (Objects.isNull(orders)) {
            throw new BusException(ErrorEnums.ORDER_EXPIRE_ERROR);
        }
        orders.setStatus(OrderStatusEnums.PAID.getCode());
        orders.setPaymentTime(new Date());
        orders.setPaymentType("2");

        //从reids删除订单
        redisTemplate.delete(orderId);
        redisTemplate.delete(orderId + "_copy");

        return orders;
    }

    @Override
    public void addRedisSeckillGoods(SeckillGoods seckillGoods) {
        if (Objects.nonNull(seckillGoods)) {
            redisTemplate.boundHashOps(RedisKey.SEC_KILL_GOODS).put(seckillGoods.getGoodsId().toString(), seckillGoods);
        }
    }
}
